use crate::vdd_polars::{self, try_proxy_sql};
use crate::{JwtClaims, WebResponseSupportTrait};
use salvo::jwt_auth::JwtAuthDepotExt;
use salvo::{handler, Depot, Request, Response};
use serde_json::Value;
use sqlparser::ast::{Statement, TableFactor};
use sqlparser::dialect::MySqlDialect;
use sqlparser::parser::Parser;
use std::sync::Arc;
use visit_dd_core::i18n_text::E_ARG_BAD;
use visit_dd_core::model::VddModel;
use visit_dd_core::query::QueryPlan;
use visit_dd_core::visit::VisitGroup;
use visit_dd_core::{i18n, Astr};
use visit_dd_db::{model_db, query_db, visit_db};

/// 解析sql,做预处理和判断
fn check_sql(sql: &str) -> Result<Vec<Astr>, Astr> {
    let ast = Parser::parse_sql(&MySqlDialect {}, sql);
    if ast.is_err() {
        return Err(Arc::from(ast.unwrap_err().to_string()));
    }
    let mut r = Vec::<Arc<str>>::new();
    let ast = ast.unwrap();
    if ast.first().is_none() {
        return Err(i18n!(1351));
    }
    match ast.first().unwrap() {
        Statement::Query(query) => {
            let body = query.body.as_select().unwrap();
            body.from.iter().for_each(|x| match x.to_owned().relation {
                TableFactor::Table { name, .. } => {
                    r.push(Arc::from(name.to_string()));
                }
                _ => {}
            });
        }
        _ => {
            return Err(i18n!(1352));
        }
    }
    Ok(r)
}

///
async fn tags_to_models(tags: &[Arc<str>]) -> Result<Vec<Arc<VddModel>>, Arc<str>> {
    let mut models = Vec::<Arc<VddModel>>::new();
    for x in tags.iter() {
        let m = model_db::m_load_by_tag(&x).await;
        if m.is_some() {
            models.push(m.unwrap());
        }
    }
    if models.len() < tags.len() {
        Err(i18n!(1353))
    } else {
        Ok(models)
    }
}

/// 检查一批模型是否是同一个用户的同一个存储配置
fn check_same_user_and_store(models: &[Arc<VddModel>]) -> Result<(), Astr> {
    if models.len() == 1 {
        return Ok(());
    }
    let last_store = models.first().unwrap().store.clone();
    for model in models {
        if (model.store.is_some() && last_store.is_none())
            || (model.store.is_none() && last_store.is_some())
        {
            return Err(i18n!(1354));
        }
        if model.store.is_some() && last_store.is_some() {
            return if model
                .store
                .clone()
                .unwrap()
                .same_as(&last_store.clone().unwrap())
            {
                Ok(())
            } else {
                Err(i18n!(1354))
            };
        }
    }
    Ok(())
}

/// 在共享的模型群执行sql语句
#[handler]
pub(crate) async fn p_run_share_sql(req: &mut Request, depot: &mut Depot, res: &mut Response) {
    let jwt: &JwtClaims = &depot.jwt_auth_data::<JwtClaims>().unwrap().claims;
    let j: Value = req.parse_body::<Value>().await.unwrap();
    let sql = j["sql"].as_str().expect(i18n!(E_ARG_BAD).as_ref());
    let r = check_sql(&sql);
    if r.is_err() {
        res.error(&r.unwrap_err());
        return;
    }
    let r = r.unwrap();
    let models = tags_to_models(&r).await;
    if models.is_err() {
        res.error(&models.unwrap_err());
        return;
    }
    let models = models.unwrap();
    let r = check_same_user_and_store(&models);
    if r.is_err() {
        res.error(&r.unwrap_err());
        return;
    }
    let mut auths = Vec::<Arc<VisitGroup>>::new();
    for x in models.iter() {
        let r = visit_db::vg_try_load_in_model_by_user(x.id, &jwt.uid).await;
        if r.is_none() {
            res.error(i18n!(1355, &x.name).as_ref());
            return;
        }
        auths.push(r.unwrap().into());
    }
    for model in models.iter() {
        vdd_polars::refresh_dataframe_cache(&model).await;
    }
    let tags = model_db::m_select_tags_to_map(true, true, None).await;
    let r = vdd_polars::try_run_share_sql(&auths, &tags, sql).await;
    if r.is_ok() {
        res.ok(&r.unwrap());
    } else {
        res.error(&r.unwrap_err());
    }
}

#[handler]
pub(crate) async fn g_my_plan(_req: &mut Request, depot: &mut Depot, res: &mut Response) {
    let jwt: &JwtClaims = &depot.jwt_auth_data::<JwtClaims>().unwrap().claims;
    let plans = visit_dd_db::query_db::select_plans(&jwt.uid).await;
    res.ok(&plans);
}

#[handler]
pub(crate) async fn p_save_plan(req: &mut Request, depot: &mut Depot, res: &mut Response) {
    let jwt: &JwtClaims = &depot.jwt_auth_data::<JwtClaims>().unwrap().claims;
    let j: Value = req.parse_body::<Value>().await.unwrap();
    let mut is_update = false;
    let plan = if let Some(plan_id) = j["id"].as_u64() {
        let mut plan2 = query_db::load_plan(&jwt.uid, plan_id).await;
        plan2.update_from_json(&j);
        is_update = true;
        plan2
    } else {
        QueryPlan::new(&jwt.uid, &j)
    };
    let r = visit_dd_db::query_db::save_plan(&plan, is_update).await;
    if r.is_ok() {
        res.ok(&r.unwrap());
    } else {
        res.error(&r.unwrap_err());
    }
}

#[handler]
pub(crate) async fn p_delete_plan(req: &mut Request, depot: &mut Depot, res: &mut Response) {
    let jwt: &JwtClaims = &depot.jwt_auth_data::<JwtClaims>().unwrap().claims;
    let j: Value = req.parse_body::<Value>().await.unwrap();
    let id = j["id"].as_u64().unwrap();
    let r = visit_dd_db::query_db::delete_plan(&jwt.uid, id).await;
    if r.is_ok() {
        res.ok(&r.unwrap());
    } else {
        res.error(&r.unwrap_err());
    }
}

/// 尝试执行一个设定的查询
#[handler]
pub(crate) async fn p_try_plan(req: &mut Request, depot: &mut Depot, res: &mut Response) {
    let jwt: &JwtClaims = &depot.jwt_auth_data::<JwtClaims>().unwrap().claims;
    let j: Value = req.parse_body::<Value>().await.unwrap();
    let plan = QueryPlan::new(&jwt.uid, &j);
    let r = check_sql(&plan.sql);
    if r.is_err() {
        res.error(&r.unwrap_err());
        return;
    }
    let r = r.unwrap();
    let models = tags_to_models(&r).await;
    if models.is_err() {
        res.error(&models.unwrap_err());
        return;
    }
    let models = models.unwrap();
    let r = check_same_user_and_store(&models);
    if r.is_err() {
        res.error(&r.unwrap_err());
        return;
    }
    for m in models.iter() {
        vdd_polars::refresh_dataframe_cache(m).await;
    }
    let r = try_proxy_sql(&jwt.uid, &plan.sql).await;
    if r.is_ok() {
        let mut sql_r = r.unwrap();
        sql_r.set_title(&plan.name);
        res.ok(&sql_r);
    } else {
        res.error(&r.unwrap_err());
    }
}

/// 删除一个查询计划
#[handler]
pub(crate) async fn p_remove_plan(req: &mut Request, depot: &mut Depot, res: &mut Response) {
    let jwt: &JwtClaims = &depot.jwt_auth_data::<JwtClaims>().unwrap().claims;
    let j: Value = req.parse_body::<Value>().await.unwrap();
    let id = j["id"].as_u64().unwrap();
    let r = visit_dd_db::query_db::delete_plan(&jwt.uid, id).await;
    if r.is_ok() {
        res.ok(&r.unwrap());
    } else {
        res.error(&r.unwrap_err());
    }
}

/// 执行一个设定的查询
#[handler]
pub(crate) async fn p_exec_plan(req: &mut Request, depot: &mut Depot, res: &mut Response) {
    let jwt: &JwtClaims = &depot.jwt_auth_data::<JwtClaims>().unwrap().claims;
    let j: Value = req.parse_body::<Value>().await.unwrap();
    let plan_id = j["id"].as_u64().unwrap();
    let plan = query_db::load_plan(&jwt.uid, plan_id).await;
    let r = check_sql(&plan.sql);
    if r.is_err() {
        res.error(&r.unwrap_err());
        return;
    }
    let r = r.unwrap();
    let models = tags_to_models(&r).await;
    if models.is_err() {
        res.error(&models.unwrap_err());
        return;
    }
    let models = models.unwrap();
    let r = check_same_user_and_store(&models);
    if r.is_err() {
        res.error(&r.unwrap_err());
        return;
    }
    for m in models.iter() {
        vdd_polars::refresh_dataframe_cache(m).await;
    }
    let r = vdd_polars::try_proxy_sql(&jwt.uid, &plan.sql).await;
    if r.is_ok() {
        let mut sql_r = r.unwrap();
        sql_r.set_title(&plan.name);
        res.ok(&sql_r);
    } else {
        res.error(&r.unwrap_err());
    }
}
